package com.reger.canal.client;

import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;

public class CanalClient implements Client {

	private static final Logger log = LoggerFactory.getLogger(CanalClient.class);
	private static final String desName = "destination";
	private volatile boolean running = false;
	private String destination;
	private CanalConnector connector;
	private MessageProcess process;
	private final Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
		public void uncaughtException(Thread t, Throwable e) {
			log.error("解析事件有一个错误 ", e);
		}
	};
	
	private final Thread thread = new Thread(new Runnable() {
		public void run() {
			process();
		}
	});

	public String getDestination() {
		return destination;
	}

	public void setDestination(String destination) {
		this.destination = destination;
	}

	public CanalConnector getConnector() {
		return connector;
	}

	public void setConnector(CanalConnector connector) {
		this.connector = connector;
	}

	public MessageProcess getProcess() {
		return process;
	}

	public void setProcess(MessageProcess process) {
		this.process = process;
	}

	@Override
	public void start() {
		Assert.notNull(connector, "连接器不能为空");
		log.info("Canal客户端启动....");
		thread.setName("main-CanalThread");
		thread.setUncaughtExceptionHandler(handler);
		thread.start();
		running = true;
	}

	@Override
	public void stop() {
		log.info("Canal客户端开始关闭 ....");
		if (!running) {
			return;
		}
		running = false;
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				log.warn("Canal客户端开始关闭 失败!", e);
			}
		}
		MDC.remove(desName);
	}

	protected void process() {
		int batchSize = 10*1024 ;
		while (running) {
			try {
				MDC.put(desName, destination);
				connector.connect();
				log.info("Connector客户端连接完毕!");
				connector.subscribe();
				log.info("Connector客户端启动完毕!");
				while (running) {
					// 获取指定数量的数据
					Message message =  connector.getWithoutAck(batchSize, 2000L, TimeUnit.MILLISECONDS);
					long batchId = message.getId();
					if (batchId == -1 || message.getEntries().isEmpty()) {
						log.debug("提交确认");
						connector.ack(batchId);
						log.debug("未拿到数据，暂停200豪秒继续订阅");
						Thread.sleep(200);
					} else {
						try {
							if (process == null) {
								log.warn("没有消息处理器，数据丢失 ....");
							} else {
								process.process(message);
							}
							log.debug("提交确认");
							connector.ack(batchId);
						} catch (Exception e) {
							log.error("处理失败, 回滚数据", e);
							connector.rollback(batchId);
							throw e;
						}
					}

				}
			} catch (InterruptedException e) {
				log.warn("线程中断...", e);
				running = false;
			} catch (Exception e) {
				log.error("线程错误!", e);
			} finally {
				log.info("connector关闭 ....");
				connector.disconnect();
				MDC.remove(desName);
			}
		}
		log.info("Canal客户端关闭 完毕！");
	}

}
